package com.gome.ss.flowcrawl.extractor;

import javax.jms.JMSException;

import org.apache.log4j.Logger;

import com.gome.ss.common.AppConfiguer;
import com.gome.ss.common.Utils;
import com.gome.ss.core.schedule.ThreadTaskBase;
import com.mongodb.BasicDBObject;
import com.mongodb.DBCollection;
import com.mongodb.DBCursor;
import com.mongodb.DBObject;

public class FlowCrawlExtractorNormalTask extends FlowCrawlExtractorTask implements ThreadTaskBase {
	private static Logger logger = Logger.getLogger(FlowCrawlExtractorNormalTask.class);
	
	public FlowCrawlExtractorNormalTask(FlowCrawlExtractor extractor) {
		super(extractor);
	}
	
	@Override
	public void execute() {
		FlowCrawlExtractorConfig config = extractor.getConfig();
		
		DBCollection collection = AppConfiguer.flowCrawlConfigMongo.getDB(AppConfiguer.FLOWCRAWL_DB_NAME)
				 .getCollection(extractor.getConfig().getName());
		
		BasicDBObject query = new BasicDBObject();
		query.append("stepID", config.getStepID());
		DBCursor cursor = collection.find(query);
		
		int totalCount = cursor.count();
		int curpos = 0;
		
		logger.info(config.getName() + " extract data count " + cursor.count());
		while(cursor.hasNext() && !needStop) {
			DBObject obj = cursor.next();
			if (obj != null) {
				try {
					putQueue(obj);
					session.commit();
				} catch (JMSException e) {
					logger.error(Utils.getTrace(e));
				}
			}
			curpos++;
			if (curpos%100 == 0)
				logger.info(config.getName() + " extract " + curpos + " has end.");
		}
		
		logger.info(config.getName() + " extract " + curpos + " end.");
		extractor.getStatus().setProgress(100);
		
		if (config.getEndCallbackLink() != null && !config.getEndCallbackLink().equals("")) {
			Utils.sendGet(config.getEndCallbackLink());
		}
		
		needStop = false;
		try {
			producer.close();
			session.close();
			connection.close();
		} catch (JMSException e) {
			logger.error(Utils.getTrace(e));
		}
		extractor.getStatus().stop();
	}
	
	@Override
	public void stop() {
		logger.info(extractor.getConfig().getName() + " extract data stop.");
		needStop = true;
	}
}